package com.gt.acc

import org.apache.spark.util.AccumulatorV2

import scala.collection.mutable
import scala.collection.mutable.Map

/**
 * 自定义类加器，实现sc
 * 1. 创建类加器
 * 2. 向spark注册
 * 3. 使用类加器
 */
class MyAccumulator extends AccumulatorV2[String,Map[String,Long]]{

  var map : Map[String, Long] = Map()

  //类加器是否为初始状态
  override def isZero: Boolean = map.isEmpty

  override def copy(): AccumulatorV2[String, Map[String, Long]] = new MyAccumulator()

  override def reset(): Unit = map.clear()

  override def add(v: String): Unit = {
//    if(map.contains(v)){
//      map.put(v,map.get(v).get+1)
//    }else{
//      map.put(v,1)
//    }
    val newCount = map.getOrElse(v, 0L) + 1
    map.put(v, newCount)
    //map.update(v, newCount)
  }

  override def merge(other: AccumulatorV2[String, Map[String, Long]]): Unit = {
//    for(kv <- other.value){
//      if(map.contains(kv._1)){
//        map.put(kv._1,map.get(kv._1).get + kv._2)
//      }else{
//        map.put(kv._1, kv._2)
//      }
//    }

    val map2: mutable.Map[String, Long] = other.value
    map2.foreach(data =>{
      val count2 = map.getOrElse(data._1,0L) +data._2
      map.update(data._1,count2)
    })
  }

  override def value: Map[String, Long] = map
}
